package com.leal.client;

import com.alibaba.fastjson.JSONArray;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.apache.log4j.Logger;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.transport.client.PreBuiltTransportClient;


import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.*;

/**
 * @Classname bigdata
 * @Description 基于es 6.1 的工具类
 * @Version 1.0
 * @Date 2024/4/1 11:12
 * @Created by leal
 */
public class ES6Client {

    public static Logger logger = Logger.getLogger(ES6Client.class);
    //使用reset api 创建client

    /**
     * es 6.1.2 直接使用transport 进行操作处理
     *
     * @return
     */
    public static TransportClient getClient() {
        System.setProperty("es.set.netty.runtime.available.processors", "false");
        return new PreBuiltTransportClient(
                Settings.builder()
                        .put("cluster.name", "elasticsearch")
                        .put("client.transport.sniff", true)
                        .build()
        ).addTransportAddress(new TransportAddress(
                new InetSocketAddress("localhost", 9300)
        ));
    }

    // 创建索引
    public static boolean createSimpleIndex(TransportClient client, String indexName) {
        return client.admin().indices().prepareCreate(indexName).get().isAcknowledged();
    }

    /**
     * 根据mapping建立索引
     *
     * @param client
     * @param indexName
     * @param esInfo    采用string 的类型有可能参数传递
     * @return
     */
    public static boolean createIndexByMapping(TransportClient client, String indexName, ESInfo esInfo) throws IOException {

        if (isExistsIndex(client, indexName)) {
            logger.info(indexName + "index already exists");
            return false;
        }

        return client.admin().indices().prepareCreate(indexName)
                .addMapping("doc", getMappingByField(esInfo.getEsMappingHashSet()))
                .execute()
                .actionGet().isAcknowledged();
    }

    // 判断索引是否存在
    public static boolean isExistsIndex(TransportClient client, String indexName) {
        return client.admin().indices().prepareExists(indexName).get().isExists();
    }

    public static boolean deleteIndex(TransportClient client, String indexName) {

        // 如果索引不存在，返回false
        if (!isExistsIndex(client, indexName)) {
            return false;
        }
        return client.admin().indices().prepareDelete(indexName).get().isAcknowledged();
    }

    public static boolean addAlias(TransportClient client, String indexName, String aliasName) {
        if (!isExistsIndex(client, indexName)) {
            return false;
        }
        return client.admin().indices().aliases(
                        new IndicesAliasesRequest()
                                .addAliasAction(
                                        IndicesAliasesRequest.AliasActions.add()
                                                .index(indexName).alias(aliasName))
                ).actionGet()
                .isAcknowledged();
    }

    public static boolean removeAlias(TransportClient client, String indexName, String aliasName) {
        // 索引是否存在 或 别名是否存在
        if (!isExistsIndex(client, indexName) || !isExistsAlias(client, indexName, aliasName)) {
            return false;
        }

        return client.admin().indices().prepareAliases().removeAlias(indexName, aliasName).get().isAcknowledged();
    }

    // 列出真实索引下的别名
    public static List<String> listAlias(TransportClient client, String indexName) {
        ImmutableOpenMap<String, List<AliasMetaData>> aliases = client.admin().indices().prepareGetAliases(
                indexName).get().getAliases();
        List<String> aliasList = new ArrayList<>();
        // 遍历获取别名
        for (ObjectObjectCursor<String, List<AliasMetaData>> alias : aliases) {
            if (alias.value.isEmpty() || alias.key.lastIndexOf("_") == -1) {
                continue;
            }

            String realAliasName = alias.value.get(0).getAlias();
            // 真实索引有别名 且别名和真实索引一致
            if (realAliasName.equals(indexName)) {
                aliasList.add(alias.key);
            }

        }
        return aliasList;
    }

    //判断真实索引是否有别名
    public static boolean isExistsAlias(TransportClient client, String indexName, String aliasName) {
        return listAlias(client, aliasName).contains(indexName);
    }

    // 获得索引的mapping
    public static Map<String, Object> getMapping(TransportClient client, String indexName) {
        return client.admin().indices().prepareGetMappings(indexName).get().mappings().get(indexName).get("doc").sourceAsMap();
    }

    //获得索引的记录数
    public static long getCount(TransportClient client, String indexName) {
        return client.prepareSearch(indexName).setTypes("doc").setSize(0).get().getHits().getTotalHits();
    }

    //更新es settings
    public static boolean updateSettings(TransportClient client, String indexName, Map<String, Object> settings) {
        return client.admin().indices().prepareUpdateSettings(indexName).setSettings(settings).get().isAcknowledged();
    }

    public static void close(TransportClient client) {
        client.close();
    }

    // 以下处理mapping

    private static void getInfoListByLine(String line, Map<String, HashSet<ESMapping>> complexFieldInfoMap,
                                          HashSet<ESMapping> fieldMappingInfoList) {
        String[] fieldInfo = line.split(",");
        if (fieldInfo.length == 4) {
            ESMapping esMapping = new ESMapping(fieldInfo[1], fieldInfo[2], "1".equals(fieldInfo[3]));
            // 处理外层字段
            if (!"0".equals(fieldInfo[0])) {
                complexFieldInfoMap.computeIfAbsent(fieldInfo[0], k -> new HashSet<>()).add(esMapping);
            } else {
                fieldMappingInfoList.add(esMapping);
            }
        } else {
            logger.info("mapping file format error: " + line);
        }

        //处理内层嵌套字段
        for (ESMapping esMapping : fieldMappingInfoList) {
            switch (esMapping.getFieldType()) {
                case "object":
                case "nested":
                    esMapping.setComplexField(complexFieldInfoMap.get(esMapping.getFieldName()));
            }
        }
    }

    public static XContentBuilder getMapping(HashSet<ESMapping> fieldInfoList, XContentBuilder mapping) throws IOException {
        //遍历fieldInfoList 判断其类型,从 properties 开始处理

        mapping.startObject("properties");
        {
            for (ESMapping esFieldMappingInfo : fieldInfoList) {
                String fieldName = esFieldMappingInfo.getFieldName().toLowerCase();
                String fieldType = esFieldMappingInfo.getFieldType().toLowerCase();
                switch (fieldType) {
                    /**
                     用于索引全文值的字段。使用文本数据类型的字段，它们会被分词，在索引之前将字符串转换为单个术语的列表(倒排索引)，
                     分词过程允许ES搜索每个全文字段中的单个单词。文本字段不用于排序，很少用于聚合（重要的术语聚合是一个例外）。
                     什幺情况适合使用text datatype，
                     只要不具备唯一性的字符串一般都可以使用text，例如：电子邮件正文，商品介绍，个人简介
                     */
                    case "string":
                        /**
                         analyzer：指明该字段用于索引时和搜索时的分析字符串的分词器（使用search_analyzer可覆盖它）。 默认为索引分析器或标准分词器
                         fielddata：指明该字段是否可以使用内存中的fielddata进行排序，聚合或脚本编写？默认值为false，可取值true或false。（排序，分组需要指定为true）
                         fields：【多数类型】text类型字段会被分词搜索，不能用于排序，而当字段既要能通过分词搜索，又要能够排序，就要设置fields为keyword类型进行聚合排序。
                         index：【是否被索引】设置该字段是否可以用于搜索。默认为true，表示可以用于搜索。
                         search_analyzer：设置在搜索时，用于分析该字段的分析器，默认是【analyzer】参数的值。
                         search_quote_analyzer：设置在遇到短语搜索时，用于分析该字段的分析器，默认是【search_analyzer】参数的值。
                         index_options：【索引选项】用于控制在索引过程中哪些信息会被写入到倒排索引中
                         - docs：只索引文档号到倒排索引中，但是并不会存储
                         - freqs：文档号和关键词的出现频率会被索引，词频用于给文档进行评分，重复词的评分会高于单个次评分
                         - positions：文档号、词频和关键词 term 的相对位置会被索引，相对位置可用于编辑距离计算和短语查询(不分词那种)
                         - offsets：文档号、词频、关键词 term 的相对位置和该词的起始字符串偏移量
                         */
                        if (esFieldMappingInfo.getParticiple()) {
                            mapping.startObject(fieldName);
                            {
                                mapping.field("type", "text");
                                mapping.field("analyzer", "ik_max_word");
                                mapping.field("search_analyzer", "ik_smart");
                                mapping.startObject("fields");
                                {
                                    mapping.startObject("keyword");
                                    {
                                        mapping.field("type", "keyword");
                                        mapping.field("ignore_above", 256);
                                    }
                                    mapping.endObject();
                                }
                                mapping.endObject();
                            }
                            mapping.endObject();
                        } else {
                            /**
                             keyword datatype，关键字数据类型，用于索引结构化内容的字段。使用keyword类型的字段，其不会被分析，给什幺值就原封不动地按照这个值索引，
                             所以关键字字段只能按其确切值进行搜索，通常用于过滤、排序和聚合。什幺情况下使用keyword datatype，具有唯一性的字符串，
                             例如：电子邮件地址、MAC地址、身份证号、状态代码…
                             eager_global_ordinals：指明该字段是否加载全局序数？默认为false，不加载。 对于经常用于术语聚合的字段，启用此功能是个好主意。
                             fields：指明能以不同的方式索引该字段相同的字符串值，例如用于搜索的一个字段和用于排序和聚合的多字段
                             ignore_above：不要索引长于此值的任何字符串。默认为2147483647，以便接受所有值
                             index：指明该字段是否可以被搜索，默认为true，表示可以被搜索
                             index_options：指定该字段应将哪些信息存储在索引中，以便用于评分。默认为docs，但也可以设置为freqs，这样可以在计算分数时考虑术语频率
                             norms：在进行查询评分时，是否需要考虑字段长度，默认为false，不考虑
                             ignore_above：默认值是256，该参数的意思是，当字段文本的长度大于指定值时，不会被索引，
                             但是会存储。即当字段文本的长度大于指定值时，聚合、全文搜索都查不到这条数据。
                             ignore_above 最大值是 32766 ，但是要根据场景来设置，比如说中文最大值 应该是设定在10922
                             */
                            mapping.startObject(fieldName);
                            {
                                mapping.field("type", "keyword");
                            }
                            mapping.endObject();
                        }
                        break;
                    case "date":
                        mapping.startObject(fieldName);
                    {
                        mapping.field("type", "date");
                        mapping.field("format", "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis");
                    }
                    mapping.endObject();
                    break;
                    // 处理复合类型
                    case "object":
                    case "nested":
                        mapping.startObject(fieldName);
                    {
                        mapping.field("type", fieldType);
                        getMapping(esFieldMappingInfo.getComplexField(), mapping);
                    }
                    mapping.endObject();
                    break;
                    default:
                        /**
                         * 数字类型
                         * ES支持的数字类型有整型数字类型，
                         * integer类型、long类型、short类型、byte类型。
                         * 浮点型数字类型 double类型、 float类型、 half_float类型、 scaled_float类型这类数据类型都是以确切值索引的，
                         * 可以使用term查询精确匹配。
                         * <p>
                         * long带符号的64位整数，最小值-263，最大值263-1
                         * integer带符号的32位整数，最小值-231，最大值231^-1
                         * short带符号的16位整数，最小值-32768，最大值32767
                         * byte带符号的8位整数，最小值-128，最小值127
                         * double双精度64位IEEE 754 浮点数
                         * float单精度32位IEEE 754 浮点数
                         * half_float半精度16位IEEE 754 浮点数
                         * scaled_float带有缩放因子的缩放类型浮点数，依靠一个long数字类型通过一个固定的(double类型)缩放因数进行缩放
                         */
                        mapping.startObject(fieldName);
                    {
                        mapping.field("type", fieldType);
                    }
                    mapping.endObject();
                }
            }
        }
        mapping.endObject();
        return mapping;
    }

    public static XContentBuilder getMappingByField(HashSet<ESMapping> esMappingHashSet) throws IOException {
        if (esMappingHashSet == null) {
            return null;
        }
        XContentBuilder mapping = XContentFactory.jsonBuilder();
        mapping.startObject();
        {
            mapping.startObject("doc");
            {
                getMapping(esMappingHashSet, mapping);
            }
            mapping.endObject();
        }
        mapping.endObject();
        return mapping;
    }

    public static HashSet<ESMapping> getMappingByFile(File file) {
        HashSet<ESMapping> esMappings = new HashSet<>();
        try {
            String line;
            BufferedReader reader = new BufferedReader(new FileReader(file));
            Map<String, HashSet<ESMapping>> complexField = new HashMap<>();
            while ((line = reader.readLine()) != null) {
                getInfoListByLine(line, complexField, esMappings);
            }
            reader.close();
        } catch (IOException e) {
            logger.error("读取文件失败", e);
        }
        logger.info("ESMapping" + esMappings);
        System.out.println("ESMapping==> " + esMappings);
        return esMappings;
    }

    public static HashSet<ESMapping> getMappingByJSONArray(JSONArray jsonArray) {
        HashSet<ESMapping> esMappings = new HashSet<>();
        if (jsonArray == null) {
            return esMappings;
        }
        Map<String, HashSet<ESMapping>> complexField = new HashMap<>();
        for (Object fieldMapping : jsonArray) {
            getInfoListByLine(fieldMapping.toString(), complexField, esMappings);
        }
        return esMappings;
    }
}
